#include "DaemonController.h"
#include "ServerCommon.h"
#include <cstdlib>
#include <cstring>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <pthread.h>
#include <time.h>
#include <cstdio>

DaemonController::DaemonController(int socket)
{
	tcpSocket = socket;
}

int DaemonController::acceptConnection()
{
	char mess[USER_MESS_LEN];
	getRequest(mess);
	if (!strncmp(mess,"request service",strlen("request service"))) {
		reply("OK");
		return 1;
	} else
		return 0;
}

int DaemonController::getApp()
{
	char mess[USER_MESS_LEN];
	char *buf;
	int data_len;
	FILE *app;
	getRequest(mess);
	appPath = generatePath("App", mess);
	getRequest(mess);
	sscanf(mess, "data %d", &data_len);
	buf = new char[data_len];
	recvData(buf, data_len);
	app = fopen(appPath.c_str(), "w");
	fwrite(buf, data_len, 1, app);
	delete buf;
	fclose(app);
	chmod(appPath.c_str(), S_IRWXU);
	reply("OK");
	return 1;
}

int DaemonController::getInput()
{
	char mess[USER_MESS_LEN];
	char *buf;
	int data_len;
	FILE *input;
	getRequest(mess);
	inputPath = generatePath("Input", mess);
	getRequest(mess);
	sscanf(mess, "data %d", &data_len);
	buf = new char[data_len];
	recvData(buf, data_len);
	input = fopen(inputPath.c_str(), "w");
	fwrite(buf, data_len, 1, input);
	delete buf;
	fclose(input);
	reply("OK");
	return 1;
}

int DaemonController::getSched()
{
	char mess[USER_MESS_LEN];
	char *buf;
	int data_len;
	FILE *sched;
	getRequest(mess);
	schedPath = generatePath("Sched", mess);
	getRequest(mess);
	sscanf(mess, "data %d", &data_len);
	buf = new char[data_len];
	recvData(buf, data_len);
	sched = fopen(schedPath.c_str(), "w");
	fwrite(buf, data_len, 1, sched);
	delete buf;
	fclose(sched);
	chmod(schedPath.c_str(), S_IRWXU);
	reply("OK");
	return 1;
}

int DaemonController::getPara()
{
	char mess[USER_MESS_LEN];
	getRequest(mess);
	sscanf(mess, "%lf %d", &CGRatio, &minNodes);
	reply("OK");
	return 1;
}

void DaemonController::getNodes()
{
	nodes = cloud.allocMachines(minNodes);
	return;
}

void DaemonController::setSched()
{
	char mess[SCHED_MESS_LEN];
	int fd_in[2];
	int fd_out[2];
	struct keyValue tmp;
	int i;
	int total;
	int pid, status;
	std::map<int, struct Machine>::iterator iter;
	pipe(fd_in);
	pipe(fd_out);
	if ((pid = fork()) == 0) {
		close(fd_in[1]);
		close(STDIN_FILENO);
		dup(fd_in[0]);
		close(fd_in[0]);
		close(fd_out[0]);
		close(STDOUT_FILENO);
		dup(fd_out[1]);
		close(fd_out[1]);
		execl(schedPath.c_str(), schedPath.c_str(), NULL);
	} else {
		close(fd_in[0]);
		close(fd_out[1]);
		sprintf(mess, "total node %d", (int) nodes.size());
		write(fd_in[1], mess, SCHED_MESS_LEN);
		for (iter=nodes.begin();iter!=nodes.end();iter++) {
			sprintf(mess, "%d %lf %d %lf",
					iter->second.CPUNum,
					iter->second.CPUPower,
					iter->second.GPUNum,
					iter->second.GPUPower);
			write(fd_in[1], mess, SCHED_MESS_LEN);
		}
		sprintf(mess, "%s", inputPath.c_str());
		write(fd_in[1], mess, SCHED_MESS_LEN);
		close(fd_in[1]);
		waitpid(pid, &status, 0);
		read(fd_out[0], mess, SCHED_MESS_LEN);
		sscanf(mess, "total %d", &total);
		for (i=0;i<total;i++) {
			read(fd_out[0], mess, SCHED_MESS_LEN);
			sscanf(mess, "node %d %d %d", &tmp.node, &tmp.key, &tmp.value);
			schedInfo.push_back(tmp);
		}
		close(fd_out[0]);
	}
	return;
}

void DaemonController::startTask()
{
	FILE *hosts;
	std::map<int, struct Machine>::iterator iter;
	std::vector<struct keyValue>::iterator iter2;
	char mess[CMM_MESS_LEN];
	int i, j;
	mpiFilePath = generatePath("mpiFile", "hosts");
	hosts = fopen(mpiFilePath.c_str(), "w");
	for (iter=nodes.begin();iter!=nodes.end();iter++)
		fprintf(hosts, "%s\n", iter->second.hostname);
	fclose(hosts);
	for (iter=nodes.begin(),i=0;iter!=nodes.end();iter++,i++) {
		pthread_mutex_lock(iter->second.end_lock);
		if (i == 0) {
			sprintf(mess, "startMaster");
			send(iter->first, mess, CMM_MESS_LEN, 0);
			sprintf(mess, "%s", mpiFilePath.c_str());
			send(iter->first, mess, CMM_MESS_LEN, 0);
			sprintf(mess, "%s", appPath.c_str());
			send(iter->first, mess, CMM_MESS_LEN, 0);
			sprintf(mess, "%s", inputPath.c_str());
			send(iter->first, mess, CMM_MESS_LEN, 0);
			sprintf(mess, "%lf", CGRatio);
			send(iter->first, mess, CMM_MESS_LEN, 0);
		} else {
			sprintf(mess, "startSlave");
			send(iter->first, mess, CMM_MESS_LEN, 0);
			sprintf(mess, "%s", inputPath.c_str());
			send(iter->first, mess, CMM_MESS_LEN, 0);
			sprintf(mess, "%lf", CGRatio);
			send(iter->first, mess, CMM_MESS_LEN, 0);
		}
		j = 0;
		for (iter2=schedInfo.begin();iter2!=schedInfo.end();iter2++)
			if (i == iter2->node)
				j++;
		sprintf(mess, "total %d", j);
		send(iter->first, mess, CMM_MESS_LEN, 0);
		for (iter2=schedInfo.begin();iter2!=schedInfo.end();iter2++) {
			if (i == iter2->node) {
				sprintf(mess, "%d %d", iter2->key, iter2->value);
				send(iter->first, mess, CMM_MESS_LEN, 0);
			}
		}
	}
	return;
}

void DaemonController::freeNodes()
{
	cloud.freeMachines(nodes);
	return;
}

int DaemonController::uploadResults()
{
	char mess[USER_MESS_LEN];
	std::string result;
	int i;
	int data_len;
	std::map<int, struct Machine>::iterator iter;
	sprintf(mess, "total node %d", (int) nodes.size());
	reply(mess);
	for (iter=nodes.begin();iter!=nodes.end();iter++) {
		pthread_mutex_lock(iter->second.end_lock);
		result = cloud.getResult(iter->first);
		data_len = result.length();
		sprintf(mess, "data %d", data_len);
		reply(mess);
		for (i=0;i<data_len;i+=USER_MESS_LEN) {
			if (i+USER_MESS_LEN >= data_len) {
				for (;i<data_len;i++)
					mess[i%USER_MESS_LEN] = result[i];
			} else {
				memcpy(mess, result.c_str()+i, USER_MESS_LEN);
			}
			reply(mess);
		}
		pthread_mutex_unlock(iter->second.end_lock);
	}
	getRequest(mess);
	if (!strncmp(mess, "OK", strlen("OK")))
		return 1;
	else
		return 0;
}

void DaemonController::cleanMess() {
	remove(appPath.c_str());
	remove(inputPath.c_str());
	remove(schedPath.c_str());
	remove(mpiFilePath.c_str());
	return;
}

DaemonController::~DaemonController()
{
	close(tcpSocket);
}

void DaemonController::getRequest(char *buf)
{
	recv(tcpSocket, buf, USER_MESS_LEN, 0);
	return;
}

void DaemonController::reply(const char *buf)
{
	char mess[USER_MESS_LEN];
	strcpy(mess, buf);
	send(tcpSocket, mess, USER_MESS_LEN, 0);
	return;
}

std::string DaemonController::generatePath(const char *prefix, const char *name)
{
	char tmp[USER_MESS_LEN];
	std::string path;
	sprintf(tmp, "_%d_%d", (int) pthread_self(), (int) time(NULL));
	path = "./";
	path += prefix;
	path += "_";
	path += name;
	path += tmp;
	return path;
}

int DaemonController::recvData(char *buf, int data_len)
{
	char mess[USER_MESS_LEN];
	int i = 0;
	while (i < data_len) {
		getRequest(mess);
		if (i + USER_MESS_LEN >= data_len) {
			for (;i<data_len;i++)
				buf[i] = mess[i%USER_MESS_LEN];
		} else {
			memcpy(buf+i, mess, USER_MESS_LEN);
			i += USER_MESS_LEN;
		}
	}
	return i;
}

CloudMachinesManager DaemonController::cloud;
